package datas

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.{DataFrame, SparkSession}


object GetData {


  def main(args: Array[String]): Unit = {


    val session: SparkSession = SparkSession.builder()
      .appName("SparkSqlRedJson")
      .master("local[*]")
      .getOrCreate()

    //
    val properties: Properties = new Properties()
    properties.setProperty("driver", "com.mysql.jdbc.Driver")
    properties.setProperty("user", "root")
    properties.setProperty("password", "123456")
    //读取数据库的数据
    val df: DataFrame = session.read.jdbc("jdbc:mysql://linux01:3306/datas?characterEncoding=UTF-8", "t_province_city", properties)
//方式一：将dataformat按rdd来做。（拆解row（t））
/*    import session.implicits._

    val ges = df.map(t => {

      val lgt = t.getAs[Double]("bdLonggitude")
      val lnt= t.getAs[Double]("  bdDimensionality")
      val province= t.getAs[String]("province")
      val city= t.getAs[String]("city")
      val district= t.getAs[String]("district")

      val geoHash = GeoHash.withCharacterPrecision(lnt,lgt,5).toBase32
      (geoHash,province,city,district)
    }).distinct()
        .toDF("geo","province","city","district")

    ges.coalesce(1).write.parquet("data_ware/datas/out_data2")
ges.show()*/

//方式二：按结构化数据处理，用自定义sql函数，用临时表查询实现
//自定义一个经纬度转geohash编码的自定义函数
  import org.apache.spark.sql.functions._
  val gps2geo=(lng:Double,lat:Double)=>{GeoHash.withCharacterPrecision(lng,lat,5).toBase32}
//将一个scala的function注册到sql引擎中，成为sql引擎中的一个“sql函数”
  session.udf.register("gps2geo",gps2geo)
    df.createTempView("area")
session.sql(
  """
    |
    |select
    |gps2geo(bdDimensionality,bdLonggitude) as geo,
    |province,
    |city,
    |district
    |from
    |area
    |
  """.stripMargin)
    //import org.apache.spark.sql.functions._
    //val gps2geo=(lng:Double,lat:Double)=>{GeoHash.withCharacterPrecision(lng,lat,5).toBase32}
/*val gps2geo_2 = udf(gps2geo)
    val res = df.select(gps2geo_2('bdDimensionality,'bdLonggitude).as("geo"),'province,'city,'district)
    */
session.close()

  }



}